/*
 * Copyright (c) 2016. 恒昌互联网运营中心.
 */

package com.credithc.eagleeye.monitor;
import com.alibaba.fastjson.JSONObject;
import com.credithc.eagleeye.agent.util.HostUtil;
import com.credithc.eagleeye.config.EagleESMonitorConfig;
import com.credithc.eagleeye.util.FastJsonConvert;
import com.credithc.eagleeye.vo.RpcTraceInfoVo;
import com.credithc.framework.es.util.ElasticSearchConfiguration;
import com.credithc.framework.es.util.ElasticSearchUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
/**
 * 描 述: <br/>
 * 作 者: zhangjin<br/>
 * 创 建：2016年06月07日<br/>
 * 版 本：v1.0.0<br>
 * <p/>
 * 历 史: (版本) 作者 时间 注释 <br/>
 */

public class EagleMonitorMgr {

    private volatile static EagleMonitorMgr instance;

    /**
     * 在eagle-springboot中收集http与redis里的鹰眼日志信息
     * 为防止eagle-eye-boot内存溢出，应为有界队列
     * 假设每条数据1kb，则4GB内存可以容纳1024*1024*4=4194304条
     */
    private volatile LinkedBlockingQueue<RpcTraceInfoVo> queue = new LinkedBlockingQueue<RpcTraceInfoVo>(102400);

    private final Logger logger = LoggerFactory.getLogger(EagleMonitorMgr.class);

    private Thread monitorThread;

    private AtomicInteger total = new AtomicInteger(0);
    private AtomicInteger fail = new AtomicInteger(0);


    private EagleMonitorMgr() {
    }

    public static EagleMonitorMgr getInstance() {
        if (instance == null) {
            synchronized (EagleMonitorMgr.class) {
                if (instance == null) {
                    EagleESMonitorConfig.loadConfig();
                    instance = new EagleMonitorMgr();
                }
            }
        }
        return instance;
    }

    /**
     *
     * <br>描 述：启动监控线程
     * <br>作 者：zj
     * <br>历 史: (版本) 作者 时间 注释
     */
    public void startMonitorThread(){
        if(monitorThread==null){
            if(EagleESMonitorConfig.EAGLE_MONITOR_ENABLE==false){
                logger.warn("注意：鹰眼监控ES推送线程未启用！");
                return;
            }
            if(StringUtils.isBlank(EagleESMonitorConfig.EAGLE_MONITOR_ES_ADDRESS)){
                logger.warn("注意：eagle-eye监控ES地址为空，将无法查看eagle-eye日志!!!!!!!!!!");
            }else{
                ElasticSearchConfiguration esConfig=new ElasticSearchConfiguration();
                esConfig.setAddress(EagleESMonitorConfig.EAGLE_MONITOR_ES_ADDRESS);
                esConfig.setShieldUser(EagleESMonitorConfig.EAGLE_MONITOR_ES_SHIELD_USER);
                esConfig.setClusterName(EagleESMonitorConfig.EAGLE_MONITOR_ES_CLUSTER_NAME);
                ElasticSearchUtil esUtil = new ElasticSearchUtil(esConfig);
                logger.info(String.format("eagle-eye监控ES地址:%s",esConfig.toString()));
                monitorThread=new Thread(new EagleESMonitorQueueTask(esUtil));
                monitorThread.setDaemon(true);
                monitorThread.start();
                logger.info(String.format("eagle-eye monitor ES thread is start!"));
            }
        }
    }
    /**
     *
     * <br>描 述：停止监控线程
     * <br>作 者：zj
     * <br>历 史: (版本) 作者 时间 注释
     */
    public void stopMonitorThread(){
        if (monitorThread != null) {
            monitorThread.interrupt();
            logger.info(String.format("eagle-eye monitor thread is stop!"));
        }
    }

    /**
     * 插入鹰眼数据到队列队尾
     *
     * @param bean
     */
    public boolean offer(final RpcTraceInfoVo bean) {
        //如果未开启,直接返回成功,不放入队列里
        if(!EagleESMonitorConfig.EAGLE_MONITOR_ENABLE){
            return true;
        }
        boolean success =false;
        try {
            success = queue.offer(bean);
            if (!success){
                fail.incrementAndGet();
            }
            total.incrementAndGet();
        } catch (Exception ex) {
            logger.error(String.format("eagle-eye offer queue exception:%s", ex.getMessage()), ex);
        }
        return success;
    }

    /**
     * 提取队首的鹰眼数据
     */
    public String poll() {
        String json = null;
        RpcTraceInfoVo bean = queue.poll();
        if(bean!=null){
            try {
                json = FastJsonConvert.convertObjectToJSONWithNullValue(bean);
            } catch (Exception ex) {
                logger.error("objec to json exception:" + ex.getMessage());
                ex.printStackTrace();
            }
        }
        return json;
    }

    public String status(){
        return getJsonStatus();
    }

    public String clear(){
        this.queue.clear();
        return getJsonStatus();
    }

    private String getJsonStatus() {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("total", total.intValue());
        jsonObject.put("fail-offer", fail.intValue());
        jsonObject.put("remain", queue.size());
        jsonObject.put("ip", HostUtil.getIp());
        return jsonObject.toJSONString();
    }
}
